
#include <netdb.h>
#include <stdlib.h>
#include <stdio.h>
#include <string.h>
#include <errno.h>
#include <strings.h>
#include <unistd.h>
#include <pthread.h>
#include <sys/types.h>
#include <sys/socket.h>
#include <arpa/inet.h>

#include "psd_conn_mx.h"
#include "psd_protocols.h"
#include "psd_logger.h"
#include "psd_config.h"
#include "psd_tpool.h"
#include "psd_modules.h"
#include "psd_settings.h"

#include "compat.h"
#include "myriexpress.h"

struct mx_listener_args {
        mx_endpoint_t endpoint;
	pthread_t accept_thread;
};

int psd_proto_mx_init();
static psdConn *psd_proto_mx_connect(const char *depot_id, psdSettings *settings);
static psdListener *psd_proto_mx_setup_listener(const char *listener_id, psdSettings *settings, int one_shot, listener_cb callback, void *arg);
static int psd_proto_mx_start_listener(psdListener *listener);
static int psd_proto_mx_stop_listener(psdListener *listener);
static void psd_proto_mx_free_listener(psdListener *listener);
static void *psd_mx_accept_handler(void *v);

static psdProtocolHandler psd_mx_handler = {
	.connect = psd_proto_mx_connect,
	.setup_listener = psd_proto_mx_setup_listener,
	.name = "mx"
};


static psdModule psd_mx_module = {
	.desc = "MX Handler Module",
	.dependencies = "mx", 			// XXX: gross hack
	.init = psd_proto_mx_init
};

psdModule *module_info() {
	return &psd_mx_module;
}


int mx_def_use_ipv6 = 0;
int mx_def_incoming_port = 5006;
int mx_def_outgoing_port = 5006;
int mx_def_incoming_recv_bufsize = 0;
int mx_def_incoming_send_bufsize = 0;
int mx_def_outgoing_recv_bufsize = 0;
int mx_def_outgoing_send_bufsize = 0;

//bu tamam
int psd_proto_mx_init() {
        mx_init();
	if (psd_add_protocol_handler(&psd_mx_handler)) {
		psd_err(0, "couldn't add protocol handler");
		goto error_exit;
	}

	return 0;

error_exit:
	return -1;
}

static psdConn *psd_proto_mx_connect(const char *hostname, psdSettings *settings) {
  int port,i;
	psdConn *ret_conn;
	int connected;
	mx_endpoint_t new_endpoint;
	mx_endpoint_addr_t rem_ep_addr;
	char *bind_interface;	
	uint64_t nic_id;
        uint32_t board_num;
        mx_return_t retval;
	mx_segment_t segment;
	mx_request_t req;
	mx_status_t status;
	uint32_t result;
	uint64_t rem_nic_id;     
	int disabled=0;

	// grab the mx settings from the given policy
	if (psd_settings_get_int_2(settings, "mx", "disabled",&disabled)!=0) {
	    disabled=0;
	}
	if(disabled)
	  return NULL;
	  
	if (psd_settings_get_int_2(settings, "mx", "port", &port) != 0) {
		port = mx_def_outgoing_port;
	}
	
	if (psd_settings_get_2(settings, "mx", "interface", &bind_interface) != 0) {
		bind_interface = NULL;
	}

	char port_str[10];
	sprintf(port_str,"%d",port);
	psd_info(1, "connecting to %s:%s", hostname, port_str);

	psd_info(10,"bind_interface : %s",bind_interface);
        if (bind_interface != NULL) {

          retval = mx_hostname_to_nic_id(bind_interface,&nic_id);
          if(retval!= MX_SUCCESS)
            {
              psd_err(0, "bind failed: hostname could not be found");
              board_num=0;
            }
          else
            {
              retval=mx_nic_id_to_board_number(nic_id,&board_num);
              if(retval!= MX_SUCCESS)
                {
                  psd_err(0, "bind failed: board is unknown");
                  board_num=0;
                }
            }
        }

	connected = 0;

	retval=mx_open_endpoint(board_num,2,0,NULL,0,&new_endpoint);
	if(retval!=MX_SUCCESS)
	  {
	    psd_err(0, "bind failed: mx_endpoint could not been opened %d",retval);
	  }
	else
	  psd_info(1, "new endpoint has been opened");
	
	//strcat(hostname,":0");

	psd_info(1, "hostname is : %s %d",hostname,strlen(hostname));
	
	char *mx_hostname;
	mx_hostname=(char *)malloc(strlen(hostname)+3);
	strcpy(mx_hostname,"");
	for(i=0;i<strlen(hostname);i++)
	  mx_hostname[i]=hostname[i];
	
	mx_hostname[strlen(hostname)]=':';
	mx_hostname[strlen(hostname)+1]=port_str[0];
	mx_hostname[strlen(hostname)+2]='\0';
	psd_info(1, "mx_hostname is : %s",mx_hostname);
	
	mx_hostname_to_nic_id(mx_hostname,&rem_nic_id);
	retval=mx_connect(new_endpoint,rem_nic_id,1,0,10,&rem_ep_addr);
	if(retval!=MX_SUCCESS)
          {
	    psd_err(0, "connect failed: mx_endpoint could not be connected %d",retval);
          }
	else
	  {
	    connected = 1;
	  }
	
	if (!connected) {
	  psd_err(0, "couldn't connect to \"%s\"", hostname);
	  goto error_exit;
	}
	else {
	  segment.segment_length=sizeof(uint32_t);
	  uint32_t temp=18;
	  segment.segment_ptr=&temp;
	  retval=mx_issend(new_endpoint,&segment,1,rem_ep_addr,1,NULL,&req);
	  if(retval!=MX_SUCCESS)
	    {
	      psd_err(0, "connect send  failed: mx_endpoint could not send connect message %d",retval);
	      goto error_exit;
	    }
	  else
	  retval=mx_wait(new_endpoint,&req,MX_INFINITE,&status,&result);	  
	}

	ret_conn = psd_conn_mx_alloc(new_endpoint,rem_ep_addr);


	if (!ret_conn) {
	  psd_err(10, "couldn't allocate socket structure");
	  goto error_exit_sd;
	}

	psd_info(1, "connected to %s", hostname);
	
	return ret_conn;

 error_exit_sd:
 error_exit:
	mx_close_endpoint(new_endpoint);
	return NULL;
}

//bu tamam
static psdListener *psd_proto_mx_setup_listener(const char *listener_id, psdSettings *settings, int one_shot, listener_cb callback, void *arg) {
	psdListener *new_listener;
	struct mx_listener_args *mx_args;

	new_listener = psd_listener_alloc();
	if (!new_listener)
		goto error_exit;

	new_listener->id = strdup(listener_id);
	if (!new_listener->id)
		goto error_exit_listener;

	new_listener->settings = psd_settings_duplicate(settings);
	if (!new_listener->settings)
		goto error_exit_listener;

	mx_args = malloc(sizeof(struct mx_listener_args));
	if (!mx_args)
		goto error_exit_listener;

	new_listener->status = LISTENER_STOPPED;
	new_listener->callback = callback;
	new_listener->arg = arg;
	new_listener->protocol = "MX";
	new_listener->proto_private = mx_args;
	new_listener->start = psd_proto_mx_start_listener;
	new_listener->stop = psd_proto_mx_stop_listener;
	new_listener->free = psd_proto_mx_free_listener;

	return new_listener;

error_exit_listener:
	psd_listener_free(new_listener);
error_exit:
	return NULL;
}

//bu tamam
static int psd_proto_mx_start_listener(psdListener *listener) {
	char *bind_interface;
	int port;
	mx_endpoint_t endpoint;
	mx_endpoint_addr_t ep_addr;
	uint64_t nic_id;
        uint32_t board_num;
        mx_return_t retval;
	struct mx_listener_args *mx_args = listener->proto_private;
	
	if (listener->status != LISTENER_STOPPED)
	  return -1;

	if (psd_settings_get_2(listener->settings, "mx", "interface", &bind_interface) != 0) {      
	  bind_interface = NULL;                                                                                                                                                                            
        } 
	
	if (bind_interface != NULL) {
	  retval = mx_hostname_to_nic_id(bind_interface,&nic_id);
	  if(retval!= MX_SUCCESS)
	    {
	      psd_err(0, "bind failed: hostname could not be found");
	      board_num=0;
	    }
	  else
	    {
	      retval=mx_nic_id_to_board_number(nic_id,&board_num);
	      if(retval!= MX_SUCCESS)
		{
		  psd_err(0, "bind failed: board is unknown");
		  board_num=0;
		}
	    }	  
        }
	
	retval=mx_open_endpoint(board_num,1,0,NULL,0,&endpoint);
	if(retval!=MX_SUCCESS)
	  {
	    psd_err(0, "bind failed: mx_endpoint could not been opened %d",retval);
	    goto error_exit;
	  }

	mx_args->endpoint = endpoint;
	
	if (pthread_create(&mx_args->accept_thread, NULL, psd_mx_accept_handler, (void *) listener) != 0) {
	  psd_err(0, "couldn't execute socket handler thread: %s", strerror(errno));
	  goto error_exit_endpoint;
	}
	
	listener->status = LISTENER_RUNNING;
	
	psd_info(0, "Started MX listener: %s", listener->id);
	
	return 0;
	
 error_exit_endpoint:
	mx_close_endpoint(endpoint);
 error_exit:
	return -1;
}

//bu tamam
static int psd_proto_mx_stop_listener(psdListener *listener) {
	struct mx_listener_args *mx_args = listener->proto_private;
	int n;

	if (listener->status != LISTENER_RUNNING) {
		psd_err(0, "tried to stop a stopped listener: %d", listener->status);
		return -1;
	}

	if ((n = pthread_cancel(mx_args->accept_thread)) < 0){
		psd_err(0, "pthread_cancel failed: %d, %s", n, strerror(errno));
		return -1;
	}

	if ((n = pthread_join(mx_args->accept_thread, NULL)) < 0){
		psd_err(0, "pthread_join failed: %d, %s", n, strerror(errno));
		return -1;
	}

	mx_close_endpoint(mx_args->endpoint);

	listener->status = LISTENER_STOPPED;

	return 0;
}

//bu tamam
static void psd_proto_mx_free_listener(psdListener *listener) {
	struct mx_listener_args *mx_args = listener->proto_private;

	if (listener->status != LISTENER_STOPPED)
		psd_proto_mx_stop_listener(listener);

	// once it is stopped, all the resources associated with the
	// listener(thread, sockets, etc) should be free, so we can just free
	// the struct
	free(mx_args);	
}

//bu da tamam gibi
static void *psd_mx_accept_handler(void *v) {
	psdConn *new_conn;
	fd_set socket_set;
	int n, i, high;
	mx_endpoint_t endpoint;
	mx_endpoint_addr_t ep_addr;
	psdListener *listener;
	struct mx_listener_args *mx_args;
	int old_state;
	uint64_t rem_nic_id;
	uint32_t rem_endpoint_id;

	listener = v;
	mx_args = listener->proto_private;

	pthread_detach(pthread_self());
	
	while(1) {
	        mx_segment_t segment;
		mx_request_t req;
		mx_return_t retval;
		mx_status_t status;
		uint32_t result;
		segment.segment_ptr=malloc(sizeof(uint64_t));
		segment.segment_length=sizeof(uint64_t);
		retval=mx_irecv(mx_args->endpoint,&segment,1,1,0xffffffffffffffff,NULL,&req);
		if(retval != MX_SUCCESS)
		  {
		    psd_err(0, "mx_irecv failed: ");
		    mx_close_endpoint(mx_args->endpoint);
		    continue;
		  }		

		retval=mx_wait(mx_args->endpoint,&req,MX_INFINITE,&status,&result);
		if(*((uint32_t *)segment.segment_ptr)!=18)
		  {
                    psd_err(0, "mx_irecv failed: ");
                    mx_close_endpoint(mx_args->endpoint);
                    continue;
                  }
		mx_decompose_endpoint_addr(status.source,&rem_nic_id,&rem_endpoint_id);
		//retval=mx_connect(mx_args->endpoint,rem_nic_id,2,0,10,&ep_addr);
		psd_info(10,"remote_nic_id : 0x%llx \t remote_endpoint_id : 0x%x",rem_nic_id,rem_endpoint_id);

		retval=mx_connect(mx_args->endpoint,
				  rem_nic_id,
				  rem_endpoint_id,
				  0,10,
				  &ep_addr);
		new_conn = psd_conn_mx_alloc(mx_args->endpoint,status.source);
		if (!new_conn) {
			psd_err(1, "psd_conn_alloc_socket() failed: %s", strerror(errno));
			mx_close_endpoint(mx_args->endpoint);
			continue;
		}

		
		new_conn->id = strdup(new_conn->description);

		psd_info(0, "new incoming connection: %s", new_conn->description);

		new_conn->settings = psd_settings_duplicate(listener->settings);

		n = listener->callback(listener, new_conn, listener->arg);
		if (n == 0 && listener->one_shot)
			goto out;

		pthread_setcancelstate(PTHREAD_CANCEL_ENABLE, &old_state);
	}

out:
	mx_close_endpoint(mx_args->endpoint);
	
	listener->status = LISTENER_STOPPED;

	return NULL;
}
